package org.nerve.data.mining.spark;


import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.SystemUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.nerve.data.mining.spark.mongo.CollectionLogModeling;

import java.nio.file.Paths;
import java.util.*;

/**
 * 完成功能：
 * 1.设置中转服务器的端口转发（所有中转器端口的请求都转发到本地端口）
 * 2.通过ssh登录到远程服务器，检查是否已经有了java环境，如果没有则不能继续执行；如果有了就上传jar包
 *      （本程序不提供java的安装功能，因为上传一个java安装包太费时间）
 * 3.启动远程服务器的jar包
 * 4.设置本地到远程服务器（运行jar的机子）之间的SSH隧道
 * 5.测试端口是否都正常
 *
 * com.jiepu.ssh.test
 * Created by zengxm on 2015/12/10 0010.
 */
public class AppConsole {

	private static SparkConf sparkConf;
	private static SparkContext sparkContext;

	public static void init(){
		if(sparkContext==null){
			if(SystemUtils.IS_OS_WINDOWS)
				System.setProperty("hadoop.home.dir", getContentPath());

			sparkConf = new SparkConf().setAppName("Nerve Spark Console").setMaster("local[*]");
			sparkContext = new SparkContext(sparkConf);

			System.out.println("Spark 环境配置完成！");
		}
	}

	public static void exit(){
		if(sparkContext!=null){
			if(!sparkContext.isStopped())
				sparkContext.stop();
		}
	}

	public static void printHelp(){
		System.out.println("-----------------------------------------");
		System.out.println("欢迎使用Nerve Spark Console演示程序");
		System.out.println("-----------------------------------------");
		System.out.println("fileCount\t对指定文件进行单词统计，如：fileCount D:/name.txt");
		System.out.println("colLog\t统计包含指定字段的数据表（在数据表采集日志中查找），参数格式：colLog fields [mongo版本>=3.2?true 或者 false]，如：");
		System.out.println("\t\tcolLog password\t\t查找包含password字段的表");
		System.out.println("\t\tcolLog email,password\t\t查找同时包含email,password字段的表");
		System.out.println("\t\tcolLog email,password true\t\t查找同时包含email,password字段的表，使用新版本的分割器（需要3.2+版本的mongo支持）");
		System.out.println("\t\tcolLog email,password true all\t\t查找同时包含email,password字段的表，使用全表扫描的方式（标准的spark filter，先将数据都加载入spark再进行筛选，效率较低，仅用于测试）");
		System.out.println("exit\texit the program");
		System.out.println("-----------------------------------------");
	}

	public static void main(String[] args) {
		init();
		//开始接受控制台输入
		Scanner scanner = new Scanner(System.in);
		String line = null;

		printHelp();
		System.out.println("please input something");

		while(!"exit".equalsIgnoreCase(line = scanner.nextLine())){
			//分解line
			String[] ks = line.split(" ");
			try{
				switch (ks[0].toLowerCase()){
					case "help": printHelp(); break;
					case "filecount":{
						FileWordCountModeling fileModeling=new FileWordCountModeling(sparkContext);
						fileModeling.compute(ks[1]);
						break;
					}
					case "collog":{
						if(StringUtils.isBlank(ks[1]))  error("请输入需要检索的字段名");
						else{
							CollectionLogModeling colModeling=new CollectionLogModeling(sparkContext);
							colModeling.addFields(ks[1].split(","));

							//如果需要mongo的版本支持
							if(ks.length>=3)
								colModeling.setVersionGTE3_2(Boolean.valueOf(ks[2]));
							//使用聚合来筛选数据
							if(ks.length>=4 && "all".equalsIgnoreCase(ks[3]))
								colModeling.computeAll();
							else
								colModeling.compute();

						}
						break;
					}

					default:
						System.out.println("please input correct command!");
						exit();
						break;
				}
			}catch (Exception e){
				e.printStackTrace();
			}
		}

		System.out.println("The Client will be exit now!");
		System.exit(0);
	}

	public static void error(String message){
		System.err.println(message);
	}

	/**
	 * 获取资源路径
	 * @return
	 */
	public static String getContentPath(){
		try{
			return Paths.get(AppConsole.class.getClassLoader().getResource("").toURI()).toString();
		}catch (Exception e){
			e.printStackTrace();
			return "";
		}
	}
}
